/**
 * 
 */
package com.pxingg.model.pubsub;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.pxingg.Global;
import com.pxingg.exception.XException;

/**
 * 消息 发布-订阅模型中的消息交换中心，是消息订阅发布者之间的桥梁.<br>
 * 负责管理所有的订阅，并在对发布的消息找到其订阅者进行处理
 * 
 * @author pxingg
 * 
 */
public class SwitchCenter
{
    /**
     * 建立一个消息的多线程执行者，发布出来的消息如果不需要立即响应的话，<br>
     * 可以提交给到线程池中执行
     * 
     * @author pxingg
     * 
     */
    protected class MessageExecutor extends Thread
    {
        /**
         * 消息执行者需要处理的消息
         */
        protected Message message = null;

        /**
         * 使用消息执行者需要处理的消息来构造消息执行者
         * 
         * @param message
         *            消息执行者需要处理的消息
         */
        public MessageExecutor(final Message message)
        {
            this.message = message;
        }

        @Override
        public void run()
        {
            processMessage(message);
        }
    }

    /**
     * 
     * 消息交换中心的线程创建工厂，替换默认实现，把线程名的命名MSC-%03d的形式
     * 
     * @author pxingg
     * 
     */
    protected class SwitchCenterThreadFactory implements ThreadFactory
    {
        /**
         * 消息交换中心的线程创建工厂，替换默认实现，把线程名的命名MSC-%03d的形式
         */
        @Override
        public Thread newThread(final Runnable r)
        {
            String name = "";
            synchronized (SwitchCenter.log)
            {
                name = String.format("MSC-%03d", ++SwitchCenter.currentThreadCounter);
            }

            return new Thread(r, name);
        }
    }

    /**
     * The log of this class
     */
    static Log log = LogFactory.getLog(SwitchCenter.class);

    /**
     * 消息交换中心线程池的大小，决定消息的并行能力
     */
    protected static int THRESD_POOL_SIZE = 32;

    /**
     * 消息交换中心线程池中的线程计数器
     */
    static int currentThreadCounter = 0;

    /**
     * 消息交换中心的核心结构， 订阅消息名和订阅者对象的映射<br>
     * 消息名的结构为：com.pxingg.message.A.B.C<br>
     * 订阅者为一个list列表,对于这消息，所有的订阅者都会接受处理这个消息
     */
    protected Map<String, List<Subscriber>> messageSubscribers = new HashMap<String, List<Subscriber>>();

    /**
     * 消息中心处理消息的线程池
     */
    protected ExecutorService threadPool = Executors.newFixedThreadPool(SwitchCenter.THRESD_POOL_SIZE, new SwitchCenterThreadFactory());

    /**
     * 增加一个订阅者对象到指定消息<br>
     * 检查目前这个消息下的所有订阅对象，如果发现这个对象已经订阅消息，那么就不需要处理，直接返回<br>
     * 如果这个对象，目前没有订阅这个消息，就在这条消息的订阅者列表中增加这个对象
     * 
     * @param pulishPoint
     *            订阅的消息
     * @param subscriber
     *            新增加的订阅者对象
     * @return true: 把订阅者新增进消息的订阅队列<br>
     *         false: 订阅者原来已经存在于消息的订阅队列
     * 
     */
    public boolean addSubscriber(final String pulishPoint, final Subscriber subscriber)
    {
        synchronized (messageSubscribers)
        {
            List<Subscriber> subscribers = messageSubscribers.get(pulishPoint);
            if (subscribers == null)
            {
                // 如果这条消息还不存在,那么就增加上这条消息
                subscribers = new LinkedList<Subscriber>();
                messageSubscribers.put(pulishPoint, subscribers);
            }

            for (final Subscriber currentSubscriber : subscribers)
            {
                // 检查目前这个消息下的所有订阅对象，如果发现这个对象已经订阅消息，那么就不需要处理，直接返回
                if (currentSubscriber == subscriber)
                {
                    SwitchCenter.log.info("subscriber " + subscriber.getName() + " have already subcribed message at " + pulishPoint);
                    return false;
                }
            }

            // 如果这个对象，目前没有订阅这个消息，就在这条节点的订阅者列表中增加这个对象
            SwitchCenter.log.info("subscriber " + subscriber.getName() + " success to subcribed message at " + pulishPoint);
            subscribers.add(subscriber);
            return true;
        }
    }

    /**
     * 获取订阅该消息的一个链表<br>
     * 如果这条消息还不存在,那么就返回一个空链表
     * 
     * @param publishPoint
     *            订阅的消息名
     * @return 订阅该消息的一个链表
     */
    public List<Subscriber> getSubscribers(final String publishPoint)
    {
        List<Subscriber> subscribers = messageSubscribers.get(publishPoint);
        if (subscribers == null)
        {
            // 如果这条消息还不存在,那么就返回一个空链表
            subscribers = new LinkedList<Subscriber>();
        }

        return subscribers;
    }

    /**
     * 由消息交换中心的线程池调用, 找到对应的订阅者，实现对消息的处理<br>
     * 如果name为com.pxingg.A.B.C<br>
     * 那么会寻找com, com.pxing, com.pxingg.A, com.pxingg.A.B, com.pxingg.A.B.C下面的订阅者， 依次执行这些订阅者的process函数，实现消息的功能
     */
    public void processMessage(final Message message)
    {
        if (SwitchCenter.log.isDebugEnabled())
        {
            SwitchCenter.log.debug("begin to process message whose content is:\n" + message);
        }

        final String[] names = StringUtils.split(message.getPublishPoint(), '.');
        if (names.length == 0)
        {
            // 消息名为空，不进行处理
            return;
        }

        String subscriberPoint = names[0];
        for (int i = 1; i < names.length; i++)
        {
            subscriberPoint += "." + names[i];
            final List<Subscriber> subscribers = Global.switchCenter.getSubscribers(subscriberPoint);

            for (final Subscriber subscriber : subscribers)
            {
                try
                {
                    if (message.isTimeOut())
                    {
                        SwitchCenter.log.warn("it is timeout and subscriber " + subscriber.getName() + "  at " + subscriberPoint + " will not process message " + message.getMessageID());
                        message.getNewResultContainer(subscriberPoint, subscriber.getName());
                        message.setCurrentResultTimeout();
                        continue;
                    }
                    if (SwitchCenter.log.isDebugEnabled())
                    {
                        SwitchCenter.log.debug("subscriber " + subscriber.getName() + " at " + subscriberPoint + " begin to process message " + message.getMessageID());
                    }
                    message.getNewResultContainer(subscriberPoint, subscriber.getName());
                    subscriber.process(subscriberPoint, message);
                }
                catch (final Exception e)
                {
                    SwitchCenter.log.error("subscriber " + subscriber.getName() + " at " + subscriberPoint + " failed to process message " + message.getMessageID());
                    SwitchCenter.log.error(XException.getStack(e));
                }
            }
        }

        message.timer.stop();
        if (SwitchCenter.log.isDebugEnabled())
        {
            SwitchCenter.log.debug(message.getSubscriberCounter() + " subscriber(s) have process message whose content is:\n" + message);
        }
    }

    /**
     * 在消息中心中发布一条消息<br>
     * 消息中心的线程队列会负责处理这条消息
     * 
     * @param message
     *            需要发布的消息
     */
    public void publish(final Message message)
    {
        try
        {
            // 把消息加入到消息中心的线程队列中
            // 消息中心的线程队列会负责处理这条消息
            if (SwitchCenter.log.isDebugEnabled())
            {
                SwitchCenter.log.debug("message is published, message contents is:\n" + message);
            }
            threadPool.execute(new MessageExecutor(message));
        }
        catch (final Exception e)
        {
            SwitchCenter.log.error("failed to publish message " + message);
            SwitchCenter.log.error(XException.getStack(e));
        }
    }

    /**
     * 在消息中心中发布一条消息<br>
     * 由自己的线程来处理这条消息，而不是依赖于消息中心的线程队列来处理这条消息
     * 
     * @param message
     *            需要发布的消息
     */
    public void publishCurrentThread(final Message message)
    {
        try
        {
            // 把消息加入到消息中心的线程队列中
            // 消息中心的线程队列会负责处理这条消息
            if (SwitchCenter.log.isDebugEnabled())
            {
                SwitchCenter.log.debug("message is published by current thread, message contents is:\n" + message);
            }
            processMessage(message);
        }
        catch (final Exception e)
        {
            SwitchCenter.log.error("failed to publish message " + message);
            SwitchCenter.log.error(XException.getStack(e));
        }
    }

    /**
     * 情况整个订阅树，删除所有的订阅者
     * 
     * @return true: 有订阅者从消息的订阅队列中删除<br>
     *         false: 没有订阅者没有订阅任何消息
     */
    public boolean removeAllSubscriber()
    {

        if (messageSubscribers.isEmpty())
        {
            SwitchCenter.log.info("there is not exists any subscriber in message switch center for removeAllSubscriber");
            return false;
        }

        synchronized (messageSubscribers)
        {
            SwitchCenter.log.info("remove all subscriber for message switch center, the content is:\n" + this);
            messageSubscribers.clear();
            return true;
        }
    }

    /**
     * 从指定消息中删除一个订阅者对象<br>
     * 如果这条消息还不存在,那么就直接返回<br>
     * 如果消息存在，就删除里面的订阅者，如果删除后，这条消息没有订阅者，那么就从消息交换中心删除这条消息
     * 
     * @param publishPoint
     *            订阅者订阅的消息名
     * @param subscriber
     *            需要删除的订阅者
     * @return true: 把订阅者从消息的订阅队列中删除<br>
     *         false: 订阅者没有订阅该消息，不需要执行删除操作
     */
    public boolean removeSubscriber(final String publishPoint, final Subscriber subscriber)
    {
        synchronized (messageSubscribers)
        {
            final List<Subscriber> subscribers = messageSubscribers.get(publishPoint);
            if (subscribers == null)
            {
                // 如果这条消息还不存在,那么就直接返回
                SwitchCenter.log.info("subscriber " + subscriber.name + " does not subscribe any message at " + publishPoint);
                return false;
            }

            // 消息存在，就删除里面的订阅者，如果删除后，这条消息没有订阅者，那么就从消息交换中心删除这条消息
            if (subscribers.remove(subscriber))
            {
                SwitchCenter.log.info("subscriber " + subscriber.name + " success to remove from publish point " + publishPoint);
                if (subscribers.size() == 0)
                {
                    SwitchCenter.log.info("publish point " + publishPoint + " does not have any subscriber, so remove it");
                    messageSubscribers.remove(publishPoint);
                }
                return true;
            }
            else
            {
                SwitchCenter.log.info("subscriber " + subscriber.name + " have not subcribed message at " + publishPoint);
                return false;
            }
        }
    }

    /**
     * 循环检查所有的消息，如果这个订阅者有订阅任何消息，就删除掉它 <br>
     * 如果删除后，这条消息没有任何订阅者，那么就从消息中心中删除这条消息
     * 
     * @param subscriber
     *            需要删除的订阅者
     * @return true: 把订阅者从消息的订阅队列中删除<br>
     *         false: 订阅者没有订阅任何消息，不需要执行删除操作
     */
    public boolean removeSubscriber(final Subscriber subscriber)
    {
        boolean isSubscribed = false;

        synchronized (messageSubscribers)
        {
            // 循环检查所有的消息，如果这个订阅者有订阅任何消息，就删除掉它
            // 如果删除后，这条消息没有任何订阅者，那么就从消息中心中删除这条消息
            final Iterator<Entry<String, List<Subscriber>>> itor = messageSubscribers.entrySet().iterator();
            while (itor.hasNext())
            {
                final Entry<String, List<Subscriber>> entry = itor.next();

                final String message = entry.getKey();
                final List<Subscriber> subscribers = entry.getValue();

                if (subscribers.remove(subscriber))
                {
                    SwitchCenter.log.info("subscriber " + subscriber.name + " success to remove from publish point " + message);
                    isSubscribed = true;

                    if (subscribers.size() == 0)
                    {
                        // 删除后，这条消息没有订阅者，那么就从消息交换中心删除这条消息
                        SwitchCenter.log.info("publish point " + message + " do not have any subscriber, so remove it");
                        // 删除掉当前的消息，注意这里不能使用messageSubscribers.remove(message);
                        // 否则会报ConcurrentModificationException
                        itor.remove();
                    }
                }
            }

            if (!isSubscribed)
            {
                // 如果没有订阅任何消息，单独记录一条日志
                SwitchCenter.log.info("subscriber " + subscriber.name + " have not subcribed any message");
            }
        }

        return isSubscribed;
    }

    /**
     * 关闭消息交换中心，如果10秒内没有全部关闭，那么就强制关闭
     */
    public void shutdown()
    {
        threadPool.shutdown();
        if (!threadPool.isShutdown())
        {
            try
            {
                // 等待10秒的安全时间
                if (!threadPool.awaitTermination(10, TimeUnit.SECONDS))
                {
                    // 取消所有正在执行得任务
                    threadPool.shutdownNow();

                    if (!threadPool.awaitTermination(10, TimeUnit.SECONDS))
                    {
                        // 再等待10秒，如果任然没有结束，就强制关闭所有的任务
                        SwitchCenter.log.error("switch center pool is not terminate");
                    }
                }
            }
            catch (final InterruptedException e)
            {
                SwitchCenter.log.info(e);
            }
        }
    }

    @Override
    public String toString()
    {
        final StringBuffer sb = new StringBuffer();

        final String format = "%-20s = %s\n";
        sb.append(String.format(format, "currentThreadCounter", SwitchCenter.currentThreadCounter));
        final Iterator<Entry<String, List<Subscriber>>> itor = messageSubscribers.entrySet().iterator();
        while (itor.hasNext())
        {
            final Entry<String, List<Subscriber>> entry = itor.next();

            final String message = entry.getKey();
            final List<Subscriber> subscribers = entry.getValue();
            for (final Subscriber subscriber : subscribers)
            {
                sb.append(String.format(format, message, subscriber.getName()));
            }
        }

        return sb.toString();

    }
}
